#ifndef __MQ_SERVER_HPP__
#define __MQ_SERVER_HPP__

#include "../mqthird/include/muduo/proto/codec.h"
#include "../mqthird/include/muduo/proto/dispatcher.h"
#include "../mqthird/include/muduo/base/Logging.h"
#include "../mqthird/include/muduo/base/Mutex.h"
#include "../mqthird/include/muduo/net/EventLoop.h"
#include "../mqthird/include/muduo/net/TcpServer.h"

#include "../mqthird/include/muduo/net/EventLoopThread.h"

#include "../mqcommon/mq_log.hpp"
#include "../mqcommon/mq_msg.pb.h"
#include "../mqcommon/mq_proto.pb.h"

#include "mq_message.hpp"
#include "mq_queue.hpp"
#include "mq_router.hpp"
#include "mq_binding.hpp"

#include "mq_connection.hpp"
#include "mq_channel.hpp"
#include "mq_consume.hpp"
#include "mq_virtualhost.hpp"

#include "../mqcommon/mq_threadpool.hpp"

//#include "mq_host.hpp"

namespace mq
{
    #define DBFILE "/meta.db"
    #define HOSTNAME "VirtualHost"
    class Server
    {
    public:
        typedef std::shared_ptr<google::protobuf::Message> MessagePtr;

        Server(int port,const std::string& basedir)
            :_server(&_baseloop,muduo::net::InetAddress("0.0.0.0",port),"Server"
                ,muduo::net::TcpServer::kReusePort)
            ,_dispatcher(std::bind(&Server::onUnknownMessage,this,std::placeholders::_1
                ,std::placeholders::_2
                ,std::placeholders::_3))
            ,_codec(std::make_shared<ProtobufCodec>(std::bind(&ProtobufDispatcher::onProtobufMessage, &_dispatcher
                ,std::placeholders::_1
                ,std::placeholders::_2
                ,std::placeholders::_3)))
            ,_consumer_manager(std::make_shared<mq::ConsumerManager>())
            ,_virtual_host(std::make_shared<mq::VirtualHost>(HOSTNAME,basedir,basedir + DBFILE))
            ,_connection_manager(std::make_shared<mq::ConnectionManager>())
            ,_threadpool(std::make_shared<threadpool>())
        {
            //针对历史消息中的所有队列，别忘了，初始化队列的消费者管理结构
            MqueueMap qm = _virtual_host->allQueues();
            for (auto &q : qm) 
            {
                _consumer_manager->initQueueConsumer(q.first);
            }
            DLOG("注册业务请求处理函数。");
            // 注册业务请求处理函数。
            _dispatcher.registerMessageCallback<mq::openChannelRequest>(
                std::bind(&Server::onOpenChannelRequest,
                this,
                std::placeholders::_1,std::placeholders::_2,std::placeholders::_3));
            _dispatcher.registerMessageCallback<mq::closeChannelRequest>(
                std::bind(&Server::onCloseChannelRequest,
                this,
                std::placeholders::_1,std::placeholders::_2,std::placeholders::_3));
            _dispatcher.registerMessageCallback<mq::declareExchangeRequest>(
                std::bind(&Server::onDeclareExchangeRequest,
                this,
                std::placeholders::_1,std::placeholders::_2,std::placeholders::_3));
            _dispatcher.registerMessageCallback<mq::deleteExchangeRequest>(
                std::bind(&Server::onDeleteExchangeRequest,
                this,
                std::placeholders::_1,std::placeholders::_2,std::placeholders::_3));
            _dispatcher.registerMessageCallback<mq::declareQueueRequest>(
                std::bind(&Server::onDeclareQueueRequest,
                this,
                std::placeholders::_1,std::placeholders::_2,std::placeholders::_3));
            _dispatcher.registerMessageCallback<mq::deleteQueueRequest>(
                std::bind(&Server::onDeleteQueueRequest,
                this,
                std::placeholders::_1,std::placeholders::_2,std::placeholders::_3));
            _dispatcher.registerMessageCallback<mq::queueBindRequest>(
                std::bind(&Server::onQueueBindRequest,
                this,
                std::placeholders::_1,std::placeholders::_2,std::placeholders::_3));
            _dispatcher.registerMessageCallback<mq::queueUnBindRequest>(
                std::bind(&Server::onQueueUnBindRequest,
                this,
                std::placeholders::_1,std::placeholders::_2,std::placeholders::_3));
            _dispatcher.registerMessageCallback<mq::basicPublishRequest>(
                std::bind(&Server::onBasicPublishRequest,
                this,
                std::placeholders::_1,std::placeholders::_2,std::placeholders::_3));
            _dispatcher.registerMessageCallback<mq::basicAckRequest>(
                std::bind(&Server::onBasicAckRequest,
                this,
                std::placeholders::_1,std::placeholders::_2,std::placeholders::_3));
            _dispatcher.registerMessageCallback<mq::basicConsumeRequest>(
                std::bind(&Server::onBasicConsumeRequest,
                this,
                std::placeholders::_1,std::placeholders::_2,std::placeholders::_3));
            _dispatcher.registerMessageCallback<mq::basicCancelRequest>(
                std::bind(&Server::onBasicCancelRequest,
                this,
                std::placeholders::_1,std::placeholders::_2,std::placeholders::_3));
            
            DLOG("注册业务请求处理函数已完成");

            _server.setConnectionCallback(
            std::bind(&Server::onConnection, this, std::placeholders::_1));
            _server.setMessageCallback(
            std::bind(&ProtobufCodec::onMessage, _codec.get(), std::placeholders::_1
                ,std::placeholders:: _2
                ,std::placeholders:: _3));
        }
        void Start()
        {
            _server.start();
            _baseloop.loop();
        }
    private:
        void onOpenChannelRequest(const muduo::net::TcpConnectionPtr& conn,
            const openChannelRequestPtr& message,
            muduo::Timestamp)
        {
            DLOG("openChannelRequest Rid:%s  Cid:%s",message->rid().c_str(),message->cid().c_str());
            Connection::Ptr mconn = _connection_manager->getConnection(conn);
            if(conn.get() == nullptr)
            {
                DLOG("打开信道，没有找到对应的connetion对象。");
                conn->shutdown();
                return ;
            }
            DLOG("打开信道");

            return mconn->openChannel(message);
        }
        void onCloseChannelRequest(const muduo::net::TcpConnectionPtr& conn,
            const closeChannelRequestPtr& message,
            muduo::Timestamp)
        {
            DLOG("CloseChannelRequest Rid:%s  Cid:%s",message->rid().c_str(),message->cid().c_str());
            Connection::Ptr mconn = _connection_manager->getConnection(conn);
            if(conn.get() == nullptr)
            {
                DLOG("关闭信道，没有找到对应的connetion对象。");
                conn->shutdown();
                return ;
            }
            DLOG("关闭信道");

            return mconn->closeChannel(message);
        }
        void onDeclareExchangeRequest(const muduo::net::TcpConnectionPtr& conn,
            const declareExchangeRequestPtr& message,
            muduo::Timestamp)
        {
            DLOG("DeclareExchangeRequest: \t Rid:%s \t  Cid:%s \t Exchange_name:%s \t exchange_type:%d",
                message->rid().c_str(),
                message->cid().c_str(),
                message->exchange_name().c_str(),
                message->exchange_type()
            );
            // 必须要用到Channel了
            Connection::Ptr mconn = _connection_manager->getConnection(conn);

            if(mconn.get() == nullptr)
            {
                DLOG("声明交换机时，没有找到对应的connetion对象。");
                conn->shutdown();
                return ;
            }

            Channel::Ptr cp = mconn->getChannel(message->cid());
            DLOG("声明交换机");

            return cp->declareExchange(message);
        }
        void onDeleteExchangeRequest(const muduo::net::TcpConnectionPtr& conn,
            const deleteExchangeRequestPtr& message,
            muduo::Timestamp)
        {
            DLOG("DeleteExchangeRequest: Rid:%s\nCid:%s\nExchange_name:%s\n",message->rid().c_str(),message->cid().c_str(),message->exchange_name().c_str());
            Connection::Ptr mconn = _connection_manager->getConnection(conn);

            if(mconn.get() == nullptr)
            {
                DLOG("删除交换机时，没有找到对应的connetion对象。");
                conn->shutdown();
                return ;
            }
            DLOG("删除交换机");

            Channel::Ptr cp = mconn->getChannel(message->cid());

            return cp->deleteExchange(message);
        }
        void onDeclareQueueRequest(const muduo::net::TcpConnectionPtr& conn,
            const declareQueueRequestPtr& message,
            muduo::Timestamp)
        {
        DLOG("DeclareQueueRequest: Rid:%s \t Cid:%s \t Queue_name:%s \t ",message->rid().c_str(),message->cid().c_str(),message->queue_name().c_str());
            Connection::Ptr mconn = _connection_manager->getConnection(conn);

            if(mconn.get() == nullptr)
            {
                DLOG("声明队列时，没有找到对应的connetion对象。");
                conn->shutdown();
                return ;
            }

            Channel::Ptr cp = mconn->getChannel(message->cid());

            DLOG("声明队列");

            return cp->declareMsgqueue(message);
        }
        void onDeleteQueueRequest(const muduo::net::TcpConnectionPtr& conn,
            const deleteQueueRequestPtr& message,
            muduo::Timestamp)
        {
            DLOG("DeleteQueueRequest: Rid:%s\nCid:%s\nQueue_name:%s\n",message->rid().c_str(),message->cid().c_str(),message->queue_name().c_str());
            Connection::Ptr mconn = _connection_manager->getConnection(conn);

            if(mconn.get() == nullptr)
            {
                DLOG("删除队列时，没有找到对应的connetion对象。");
                conn->shutdown();
                return ;
            }

            Channel::Ptr cp = mconn->getChannel(message->cid());
            DLOG("删除队列");

            return cp->deleteMsgqueue(message);
        }
        void onQueueBindRequest(const muduo::net::TcpConnectionPtr& conn,
            const queueBindRequestPtr& message,
            muduo::Timestamp)
        {
            DLOG("QueueBindRequest: Rid:%s\nCid:%s\nExchange_name:%s\nQueue_name:%s\nbind_key:%s\n",message->rid().c_str(),message->cid().c_str(),message->exchange_name().c_str(),message->queue_name().c_str(),message->bind_key().c_str());
            Connection::Ptr mconn = _connection_manager->getConnection(conn);

            if(mconn.get() == nullptr)
            {
                DLOG("绑定队列时，没有找到对应的connetion对象。");
                conn->shutdown();
                return ;
            }

            Channel::Ptr cp = mconn->getChannel(message->cid());
            DLOG("绑定队列");

            return cp->queueBind(message);
        }
        void onQueueUnBindRequest(const muduo::net::TcpConnectionPtr& conn,
            const queueUnBindRequestPtr& message,
            muduo::Timestamp)
        {
            DLOG("QueueBindRequest: Rid:%s\nCid:%s\nExchange_name:%s\nQueue_name:%s\n",message->rid().c_str(),message->cid().c_str(),message->exchange_name().c_str(),message->queue_name().c_str());
            Connection::Ptr mconn = _connection_manager->getConnection(conn);

            if(mconn.get() == nullptr)
            {
                DLOG("解除绑定队列时，没有找到对应的connetion对象。");
                conn->shutdown();
                return ;
            }

            Channel::Ptr cp = mconn->getChannel(message->cid());
            DLOG("解除绑定队列");

            return cp->queueUnbind(message);
        }
        void onBasicPublishRequest(const muduo::net::TcpConnectionPtr& conn,
            const basicPublishRequestPtr& message,
            muduo::Timestamp)
        {
            // DLOG("消息发布")
            DLOG("onBasicPublishRequest: Rid:%s \t Cid:%s \t Exchange_name:%s \t Body:%s \t ",message->rid().c_str(),message->cid().c_str(),message->exchange_name().c_str(),message->body().c_str());
            Connection::Ptr mconn = _connection_manager->getConnection(conn);

            if(mconn.get() == nullptr)
            {
                DLOG("解除绑定队列时，没有找到对应的connetion对象。");
                conn->shutdown();
                return ;
            }

            Channel::Ptr cp = mconn->getChannel(message->cid());

            if (cp.get() == nullptr) 
            {
                DLOG("发布消息时，没有找到信道！");

                return ;
            }
            cp->basicPublish(message);
            
            return ;
        }
        //消息确认
        void onBasicAckRequest(const muduo::net::TcpConnectionPtr& conn,
            const basicAckRequestPtr& message,
            muduo::Timestamp)
        {
            DLOG("onBasicAckRequest: Rid:%s\nCid:%s\nQueue_name:%s\nMessage_id:%s\n",message->rid().c_str(),message->cid().c_str(),message->queue_name().c_str(),message->message_id().c_str());
            Connection::Ptr mconn = _connection_manager->getConnection(conn);

            if(mconn.get() == nullptr)
            {
                DLOG("解除绑定队列时，没有找到对应的connetion对象。");
                conn->shutdown();
                return ;
            }

            Channel::Ptr cp = mconn->getChannel(message->cid());

            DLOG("确认信息，找到chanel信道了");

            return cp->basicAck(message);
        }
        //队列消息订阅 
        void onBasicConsumeRequest(const muduo::net::TcpConnectionPtr& conn, 
            const basicConsumeRequestPtr& message, 
            muduo::Timestamp) 
        { 
            DLOG("创建消费者。");
            DLOG("onBasicConsumeRequest: Rid:%s \t Cid:%s \t consume_tag:%s \t queue_name:%s",
                message->rid().c_str(),
                message->cid().c_str(),
                message->consumer_tag().c_str(),
                message->queue_name().c_str());
            Connection::Ptr mconn = _connection_manager->getConnection(conn); 
            if (mconn.get() == nullptr) 
            { 
                DLOG("队列消息订阅时，没有找到连接对应的Connection 对象！"); 
                conn->shutdown(); 

                return ; 
            }
             
            Channel::Ptr cp = mconn->getChannel(message->cid()); 

            if (cp.get() == nullptr) 
            { 
                DLOG("队列消息订阅时，没有找到信道！");
            
                return ; 
            } 

            DLOG("订阅信息，找到channel信道了");

            return cp->basicConsume(message); 
        } 
        // 取消队列。
        void onBasicCancelRequest(const muduo::net::TcpConnectionPtr& conn,
            const basicCancelRequestPtr& message,
            muduo::Timestamp)
        {
            DLOG("onBasicCancelRequest: Rid:%s\nCid:%s\nconsume_tag:%s\nqueue_name:%s\n",message->rid().c_str(),message->cid().c_str(),message->consumer_tag().c_str(),message->queue_name().c_str());

            Connection::Ptr mconn = _connection_manager->getConnection(conn);

            if(mconn.get() == nullptr)
            {
                DLOG("解除绑定队列时，没有找到对应的connetion对象。");
                conn->shutdown();
                return ;
            }

            Channel::Ptr cp = mconn->getChannel(message->cid());

            DLOG("取消队列，找到信道了。")

            return cp->basicCancel(message);
        }
        void onUnknownMessage(const muduo::net::TcpConnectionPtr& conn
            ,const MessagePtr& message,muduo::Timestamp)
        {
            LOG_INFO << "onUnknownMessage: " << message->GetTypeName();
            conn->shutdown();
        }
        void onConnection(const muduo::net::TcpConnectionPtr& conn)
        {
            if(conn->connected() == true)
            {
                LOG_INFO << "新链接已建立" ;
                _connection_manager->newConnection(_virtual_host,_consumer_manager,_codec,conn,_threadpool);
            }
            else
            {
                LOG_INFO << "链接即将关闭" ;
                _connection_manager->delConnection(conn);
            }
        }     
    private:
        muduo::net::EventLoop _baseloop;
        muduo::net::TcpServer _server;
        ProtobufDispatcher _dispatcher;
        ProtobufCodecPtr _codec;
        VirtualHost::Ptr _virtual_host;
        ConsumerManager::Ptr _consumer_manager;
        ConnectionManager::Ptr _connection_manager;
        threadpool::Ptr _threadpool;
    };
}

#endif